package com.alison.tableapisql.chapter1_tableapiandsql.connector;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.Elasticsearch;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.table.types.DataType;

/**
 * @Author alison
 * @Date 2024/4/12 10:48
 * @Version 1.0
 * @Description
 */
public class E1_TableES {

    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String filePath = "D:\\workspace\\lab\\learnbigdata\\learnflink\\flink-datastream\\src\\main\\resources\\tableapi\\E1.txt";
        tableEnv.connect(new FileSystem().path(filePath))
                .withFormat(new Csv())
                .withSchema(new Schema()
                        .field("id", DataTypes.STRING())
                        .field("timestamp", DataTypes.BIGINT())
                        .field("temp", DataTypes.DOUBLE())
                )
                .createTemporaryTable("inputTable");

        tableEnv.connect(new Elasticsearch()
                .version("7")
                .host("localhost", 9200, "http")
                .index("sensor")
                .documentType("temp"))
                .inUpsertMode()
                .withFormat(new Json())
                .withSchema(new Schema()
                        .field("id", DataTypes.STRING())
                        .field("count", DataTypes.DOUBLE()))
                .createTemporaryTable("esOutputTable");

        Table aggResultTable = tableEnv.sqlQuery("select id, temp from inputTable where id = 'senosr_6'");
//        aggResultTable.insertInto("esOutputTable");
        aggResultTable.executeInsert("esOutputTable");
//        tableEnv.execute("");
//        env.execute();
    }
}
